package cn.victory;

import org.junit.Test;
import org.redisson.Redisson;
import org.redisson.api.*;
import org.redisson.api.listener.MessageListener;
import org.redisson.codec.SerializationCodec;
import org.redisson.config.Config;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @ Description TODO
 * @ Author victory
 * @ Date 2022/7/11
 **/
public class RedissonDemo {


    /**
     * 通用对象桶，可以用来存放任类型的对象
     */
    @Test
    public void RedissonBucket(){
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作对象桶来存储对象(同步)====================
        RBucket<Object> bucket = redissonClient.getBucket("name");
        //设置值为victory，过期时间为3小时
        bucket.set("victory",30, TimeUnit.HOURS);
        Object value = bucket.get();
        System.out.println(value);
        //通过key取value值
        Object name = redissonClient.getBucket("name").get();
        System.out.println(name);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }


    /**
     * 二进制流
     * 提供了InputStream接口和OutputStream接口的实现
     */
    @Test
    public void RedissonStream() throws IOException {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作流来存储对象====================
        RBinaryStream stream = redissonClient.getBinaryStream("stream");
        stream.set("name is ".getBytes());
        OutputStream outputStream = stream.getOutputStream();

        outputStream.write("victory".getBytes());
        InputStream inputStream = stream.getInputStream();

        ByteArrayOutputStream result = new ByteArrayOutputStream();
        byte[] bytes = new byte[1024];
        int length;
        while ((length = inputStream.read(bytes)) != -1) {
            result.write(bytes, 0, length);
        }
        System.out.println(result);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }




    /**
     * list
     * Redisson操作list
     */
    @Test
    public void list() {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作list====================
        RList<String> list = redissonClient.getList("list");
        list.add("victory1");
        list.add("victory2");
        System.out.println(list);
        //取值
        List<Object> list1 = redissonClient.getList("list").get();
        System.out.println(list1);
        //移除索引0位置元素
        list.remove(0);
        System.out.println(list);
        //通过key取value值
        List<Object> list2 = redissonClient.getList("list").get();
        System.out.println(list2);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }


    /**
     * set
     * Redisson操作set
     * 有序结合sorted sets用法完全一致，不单独举例
     */
    @Test
    public void set() {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作list====================
        RSet<Object> set = redissonClient.getSet("set");
        set.add("victory1");
        set.add("victory2");
        System.out.println(set);
        //通过key取value值
        RSet<Object> set1 = redissonClient.getSet("set");
        System.out.println(set1);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }


    /**
     * map
     * Redisson操作map
     * Redisson将Redis中的字符串数据结构封装成了RMap，就是原本redis中的string类型
     */
    @Test
    public void map() {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作list====================
        RMap<Object, Object> map = redissonClient.getMap("map");
        map.put("name1","victory1");
        map.put("name2","victory2");
        map.forEach((key,value)->{
            System.out.println("key = "+key+" ,value = "+ value);
        });

        //通过key取value值
        Object o = redissonClient.getMap("map").get("name1");
        System.out.println(o);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }



    /**
     * 队列
     * Redisson操作queue
     */
    @Test
    public void queue() {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作queue====================
        RQueue<String> queue = redissonClient.getQueue("queue");
        //存值
        queue.add("victory1");
        queue.add("victory2");
        //取值
        String item = queue.poll();
        System.out.println(item);

        //
        RQueue<Object> queue1 = redissonClient.getQueue("queue");
        System.out.println(queue1);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }



    /**
     * 限流器
     * Redisson操作rateLimiter
     */
    @Test
    public void rateLimiter() throws InterruptedException {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作rateLimiter====================
        RRateLimiter rateLimiter = redissonClient.getRateLimiter("rateLimiter");
        //创建限流器，最大流速:每1秒钟产生20个令牌
        rateLimiter.trySetRate(RateType.OVERALL, 20, 1, RateIntervalUnit.SECONDS);
        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                int i = 0;
                @Override
                public void run() {
                    while(true) {
                        rateLimiter.acquire(1);
                        System.out.println(Thread.currentThread() + "-" + System.currentTimeMillis() + "-" + i++);
                    }
                }
            }).start();
        }
        //等待执行完成,不设置等待可能出现还未执行完成客户端就关闭的情况
        Thread.sleep(5000);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }



    /**
     * 可重入锁
     * Redisson操作RLock
     */
    @Test
    public void lock() throws InterruptedException {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);

        //====================操作lock====================
        RLock lock = redissonClient.getLock("lock");
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread() + "-" + System.currentTimeMillis() + "-" + "获取了锁");
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }).start();
        }
        //等待执行完成,不设置等待可能出现还未执行完成客户端就关闭的情况
        Thread.sleep(5000);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }




    /**
    *  发布订阅操作
    *  Redisson操作RTopic执行发布订阅操作
    **/
    @Test
    public void topicPublisherAndSubscriber() throws InterruptedException {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);
        RedissonClient redissonClient1 = Redisson.create(config);

        //====================操作topic执行发布操作====================
        RTopic topic1 = redissonClient.getTopic("topic",new SerializationCodec());
        topic1.publish(new Message(1L,"victory",18));

        //====================操作topic执行订阅操作====================
        Thread.sleep(5000);
        RTopic topic = redissonClient1.getTopic("topic", new SerializationCodec());
        topic.addListener(Message.class, new MessageListener<Message>() {
            @Override
            public void onMessage(CharSequence channel, Message msg) {
                System.out.println("onMessage:=========" + channel + "; Thread:========= " + Thread.currentThread());
                System.out.println(" name : " + msg.getName() + " age : " + msg.getAge());
                LoggerFactory.getLogger(RedissonDemo.class).info("Redisson接收到消息",msg);
            }
        });


        //====================关闭客户端====================
        redissonClient.shutdown();
        redissonClient1.shutdown();
    }



    /**
     * Redisson利用Redis实现了Java分布式布隆过滤器（Bloom Filter）
     * 作用：在缓存层前添加布隆过滤器，常用于高并发场景下应对缓存穿透问题
     * 布隆过滤器是一个非常长的二进制向量和一系列随机哈希函数的组合，可用于检索一个元素是否存在；
     * 建立一个二进制向量，所有位设置0；
     选择K个散列函数，用于对元素进行K次散列，计算向量的位下标；
     添加元素：将K个散列函数作用于该元素，生成K个值作为位下标，将向量的对应位设置为1；
     检索元素：将K个散列函数作用于该元素，生成K个值作为位下标，若向量的对应位都是1，则说明该元素可能存在；否则，该元素肯定不存在；
     */
    @Test
    public void bloomFilter() {
        //====================创建Redisson客户端====================
        Config config = new Config();
        //config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);


        //====================操作布隆过滤器====================
        RBloomFilter<String> bloomFilter = redissonClient.getBloomFilter("bloom-filter");
        // 初始化布隆过滤器，初始化预期插入的数据量为200，期望误差率为0.01
        bloomFilter.tryInit(200, 0.01);
        //插入数据
        bloomFilter.add("丁胜利");
        bloomFilter.add("丁向前");
        bloomFilter.add("胜利丁");
        //判断是否包含
        boolean victory = bloomFilter.contains("丁胜利");
        boolean forward = bloomFilter.contains("向前丁");
        System.out.println(victory);
        System.out.println(forward);

        //====================关闭客户端====================
        redissonClient.shutdown();
    }
}
